package com.hisense.ovcloud.mqtt.monitor.service;

import com.alibaba.fastjson.JSON;
import com.hisense.ovcloud.mqtt.monitor.config.Constants;
import com.hisense.ovcloud.mqtt.monitor.config.MqttClientConfig;
import com.hisense.ovcloud.mqtt.monitor.dto.TestResult;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.mqtt.MqttClient;
import io.vertx.mqtt.MqttClientOptions;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.HmacUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;

@Service
@Slf4j
public class MQTTClientConnector {

    private final MqttClientConfig mqttClientConfig;

    private final MqttClients mqttClients;

    public final static AtomicReference<MqttClient> client = new AtomicReference<>();
    private final AtomicReference<TestResult> testResult = new AtomicReference<>(new TestResult());

    public MQTTClientConnector(MqttClientConfig mqttClientConfig, MqttClients mqttClients) {
        this.mqttClientConfig = mqttClientConfig;
        this.mqttClients = mqttClients;
    }

    @Bean("mqttCustomClient")
    public MqttClient mqttCustomClient() {
        return initClient();
    }

    public void reconnect(int checkIntervalSeconds) {
        Flux.interval(Duration.ofSeconds(checkIntervalSeconds))
                .subscribe(t -> {
                    if ((client.get() == null || !client.get().isConnected())) {
                        if(client.get() != null){
                            client.get().disconnect();
                        }
                        log.error("business is not connecting to emqx, will try to reconnect.");
                        try {
                            client.set(initClient());
                        } catch (Exception e) {
                            log.error("failed to reconnect to emqx and re-subscribe, will retry in {}s", checkIntervalSeconds);
                        }
                    }
                });
    }

    private MqttClient initAndSubscribe(int port, String host, MqttClientOptions options) {
        io.vertx.mqtt.MqttClient newClient = io.vertx.mqtt.MqttClient.create(Vertx.vertx(), options);
        newClient.connect(port, host, ch -> {
            if (ch.succeeded()) {
                log.info("Connected to emqx: {}", host + ":" + port);
                testResult.get().setConnectR(true);

                newClient.publishHandler(s -> {
                    testResult.get().setConnectR(true);
                    log.info("success to get set msg1 {} from topic {}", s.payload().toString(), s.topicName());
                    if(s.topicName().contains("set")){
                        testResult.get().setSetR(true);
                    }else{
                        testResult.get().setUploadR(true);
                    }
                    log.info("2. testResult1: {}", JSON.toJSONString(testResult.get()));
                }).subscribe(Map.of(String.format(Constants.MQTT.Topics.SET_R_TEMPLATE, mqttClientConfig.getDomain(),
                                mqttClientConfig.getSubDomain(),
                                mqttClientConfig.getWifiId()),
                        MqttQoS.EXACTLY_ONCE.value(), String.format(Constants.MQTT.Topics.UPDATE_CON_REPLY_TEMPLATE, mqttClientConfig.getDomain(),
                                mqttClientConfig.getSubDomain(),
                                mqttClientConfig.getWifiId()),
                        MqttQoS.EXACTLY_ONCE.value()));

            } else {
                log.error("Failed to connect to emqx: {}, reason: {}", host + ":" + port, ch.cause().toString());
            }
        });

        newClient.subscribeCompletionHandler(h -> {
            log.info("Subscribe complete, levels [{}]", h.grantedQoSLevels());
        });

        newClient.exceptionHandler(e -> {
            log.error("mqtt client encounterred errors: {}", e.getMessage());
        });

        newClient.closeHandler(e -> {
            log.info("close the connection now.");
        });

        return newClient;
    }

    private static String calcutePassword(String username, String timestamp, String preKey) {
        return HmacUtils.hmacMd5Hex(preKey.getBytes(StandardCharsets.UTF_8), String.format("clientId%stimestamp%s", username, timestamp).getBytes(StandardCharsets.UTF_8));
    }

    private boolean send(String topic, Buffer payload) {
        client.get().publish(
                topic,
                payload,
                MqttQoS.AT_LEAST_ONCE,
                false,
                false,
                z -> log.info("publish msg: {} to topic: {}", payload, topic));
        return true;

    }

    public MqttClient initClient() {
        if (client.get() != null && client.get().isConnected()) {
            return client.get();
        }

        String timestamp = String.valueOf(System.currentTimeMillis() / 1000);
        String clientId = String.format("%s/%s/%s_1_1_%s",
                mqttClientConfig.getDomain(),
                mqttClientConfig.getSubDomain(),
                mqttClientConfig.getWifiId(),
                timestamp
        );
        String username = String.format("%s/%s/%s",
                mqttClientConfig.getDomain(),
                mqttClientConfig.getSubDomain(),
                mqttClientConfig.getWifiId()
        );
        String password = calcutePassword(username, timestamp, mqttClientConfig.getPrekey());

        log.info("begin to connect via clientid: {}, username:{}, password: {}", clientId, username, password);
        MqttClientOptions options = new MqttClientOptions()
                .setMaxMessageSize(100_000_000_0)
                .setCleanSession(true)
                .setSsl(true)
                .setTrustAll(true)
                .setClientId(clientId)
                .setUsername(username)
                .setPassword(password);
        options.setConnectTimeout(mqttClientConfig.getConnectTimeout());
        client.set(initAndSubscribe(mqttClientConfig.getSslPort(), mqttClientConfig.getBrokerHost(), options));

        return client.get();
    }

    @SneakyThrows
    public Mono<ServerResponse> monitor(ServerRequest req) {
        return ServerResponse.ok().body(test(), String.class);
    }

    @SneakyThrows
    public Mono<String> test() {
        testResult.set(new TestResult());
        try {
            Thread.sleep(mqttClientConfig.getWaitTime());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        return Mono.zip(
                Mono.just(send(String.format(Constants.MQTT.Topics.UPDATE_CON_TEMPLATE, mqttClientConfig.getDomain(),
                        mqttClientConfig.getSubDomain(),
                        mqttClientConfig.getWifiId()), Buffer.buffer(mqttClientConfig.getUpdateConMsg()))),
                mqttClients.sendCmd(Integer.parseInt(mqttClientConfig.getDomain()),
                        Integer.parseInt(mqttClientConfig.getSubDomain()),
                        mqttClientConfig.getWifiId(), mqttClientConfig.getCmdMsg())
        ).zipWhen(res2 -> {
            try {
                Thread.sleep(mqttClientConfig.getWaitTime());
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            return Mono.just(formatTestResult(testResult.get()));
        }, (x, y) -> y);
    }

    private int booleanToInt(boolean flag) {
        return flag ? 1 : 0;
    }

    private String formatTestResult(TestResult testResult) {
        return String.format("connectR %d\nuploadR %d\nsetR %d",
                booleanToInt(testResult.isConnectR()),
                booleanToInt(testResult.isUploadR()),
                booleanToInt(testResult.isSetR()));
    }

    public static void main(String[] args) {
        System.out.println(calcutePassword("16/218/86505510007000400560000012e9dd016c02_1_1_1713414620", "1713414620", "96bd59b268a2db10be50576b17c7c4f6"));
    }
}
